package qy.qyalgotrader.wind.barfeed;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
import java.nio.file.Path;
import java.time.LocalDateTime;
import java.time.ZonedDateTime;
import java.time.Duration;
import java.time.LocalDate;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Predicate;

import com.zeroc.Ice.ACMHeartbeat;
import com.zeroc.Ice.Current;
import com.zeroc.Ice.ObjectAdapter;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.math3.stat.StatUtils;
import org.apache.commons.math3.stat.descriptive.moment.Mean;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.math.DoubleMath;
import com.zeroc.Ice.ObjectPrx;

import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
import qy.jalgotrade.bar.Bar;
import qy.jalgotrade.bar.Bar.Frequency;
import qy.jalgotrade.bar.Bars;
import qy.jalgotrade.bar.BasicBar;
import qy.jalgotrade.barfeed.BaseBarFeed;
import qy.jalgotrade.utils.CommonUtils;
import qy.jalgotrade.utils.resamplebase.TimeRange;
import qy.qyalgotrader.utils.CommonDataUtils;
import qy.qyalgotrader.utils.Constants;
import qy.qyalgotrader.utils.Helper;
import qy.qyalgotrader.utils.Services;
import qy.qyalgotrader.utils.apiservice.wind.WapiPrx;
import qy.qyalgotrader.utils.apiservice.wind.WindData;
import qy.qyalgotrader.utils.apiservice.wind.WsqCallBack;
import qy.qyalgotrader.utils.apiservice.wind.WsqCallBackPrx;
import qy.qyalgotrader.wind.WindDataUtils;
import tech.tablesaw.api.ColumnType;
import tech.tablesaw.api.DateTimeColumn;
import tech.tablesaw.api.DoubleColumn;
import tech.tablesaw.api.Row;
import tech.tablesaw.api.Table;
import tech.tablesaw.columns.Column;
import tech.tablesaw.io.csv.CsvReadOptions;

/**
 * <p>
 * 使用 Wind 实现的实时行情 BaseBarFeed.
 * </p>
 * <p>
 * 要求 identifiers 的交易时间段一致 (上午开盘, 盘休 (中国商品期货), 上午收盘, 下午开盘 必须一致, 收盘可不一样 (多品种中国商品期货)).
 * </p>
 * 
 * @author c-geo
 *
 */
public class WindLiveFeed extends BaseBarFeed {

	private static final Logger logger = LoggerFactory.getLogger(WindLiveFeed.class);

	/**
	 * w.start() 超时时间:
	 */
	private static final long WAPI_START_TIMEOUT_MS = 5000;

	/**
	 * timeout: 10, timeunit: {@link TimeUnit#MILLISECONDS MILLISECONDS)
	 */
	private static final Pair<Long, TimeUnit> QUEUE_POLL_TIMEOUT = Pair.of(10L, TimeUnit.MILLISECONDS);

	/**
	 * 中国商品行情 分笔成交 采样间隔: 1s
	 */
	public static final Pair<Long, TimeUnit> WIND_FUTURES_INQUERY_PERIOD_DEFAULT = Pair
	        .of((long) Frequency.SECOND.getValue(), TimeUnit.SECONDS); // WindQuantApi: wsq() read period, default is 1s

	/**
	 * 分笔成交 采样间隔: 3s (中国 A 股 Level I)
	 */
	public static final Pair<Long, TimeUnit> WIND_INQUERY_PERIOD_LEVEL_I = Pair
	        .of((long) Frequency.LEVEL_I_SNAPSHOT_STOCK_A.getValue(), TimeUnit.SECONDS);

	/**
	 * 分笔成交 采样间隔: 6s (低负载)
	 */
	public static final Pair<Long, TimeUnit> WIND_INQUERY_PERIOD_LOW_LOAD = Pair
	        .of((long) Frequency.TICK_MONITOR_HIGH_LOAD.getValue(), TimeUnit.SECONDS);

	/**
	 *
	 */
	public static final List<String> WSQ_FIELDS = new ArrayList<>(Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.rowKeySet());

	/**
	 * wsq() 字段: Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES
	 */
	public static final String WSQ_FIELDS_STRING = String.join(",",
			Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.rowKeySet());

	/**
	 * 
	 * @author c-geo
	 *
	 */
	private enum EventType {
		ON_BARS
	}

	/**
	 * 
	 * @author c-geo
	 *
	 */
	public static class TickDataSeries {

		private final List<LocalDateTime> __dateTimes;

		private final DoubleArrayList __priceDS;

		private final DoubleArrayList __volumeDS;

		private final DoubleArrayList __amountDS;

		private final Map<String, DoubleArrayList> __extraDSs;

		/**
		 * 
		 */
		public TickDataSeries() {

			__priceDS = new DoubleArrayList();
			__volumeDS = new DoubleArrayList();
			__amountDS = new DoubleArrayList();
			__dateTimes = new ArrayList<>();
			__extraDSs = new HashMap<>();
		}

		/**
		 * 
		 */
		public void reset() {

			__priceDS.clear();
			__volumeDS.clear();
			__amountDS.clear();
			__dateTimes.clear();
			__extraDSs.forEach((extraField, dal) -> {
				dal.clear();
			});
		}

		/**
		 * 
		 * @return __dateTimes
		 */
		public List<LocalDateTime> getDateTimes() {

			return __dateTimes;
		}

		/**
		 * 
		 * @return __priceDS
		 */
		public DoubleArrayList getPriceDS() {

			return __priceDS;
		}

		/**
		 * 
		 * @return __volumeDS
		 */
		public DoubleArrayList getVolumeDS() {

			return __volumeDS;
		}

		/**
		 * 
		 * @return __amountDS
		 */
		public DoubleArrayList getAmountDS() {

			return __amountDS;
		}

		/**
		 * 
		 * @return __extraDSs
		 */
		public Map<String, DoubleArrayList> getExtraDSs() {

			return __extraDSs;
		}

		/**
		 * 
		 * @param extraField extraField
		 * @return __extraDSs.get(extraField)
		 */
		public DoubleArrayList getExtraDS(String extraField) {

			return __extraDSs.get(extraField);
		}

		/**
		 * 
		 * @return __priceDS.size() == 0
		 */
		public boolean empty() {

			return __priceDS.size() == 0;
		}

		/**
		 * 
		 * @return __priceDS.size()
		 */
		public int size() {

			return __priceDS.size();
		}

		/**
		 * 
		 * @param dateTime dateTime
		 * @param price price
		 * @param volume volume
		 * @param amount amount
		 */
		public void appendTick(LocalDateTime dateTime, double price, double volume, double amount) {

			appendTick(dateTime, price, volume, amount, Collections.emptyMap());
		}

		/**
		 * 
		 * @param dateTime dateTime
		 * @param price price
		 * @param volume volume
		 * @param amount amount
		 * @param extras extras
		 */
		public void appendTick(LocalDateTime dateTime, double price, double volume, double amount,
		        Map<String, Double> extras) {

			__dateTimes.add(dateTime);
			__priceDS.add(price);
			__volumeDS.add(volume);
			__amountDS.add(amount);

			for (String extraField : extras.keySet()) {
				if (!__extraDSs.containsKey(extraField)) {
					__extraDSs.put(extraField, new DoubleArrayList());
				}
				__extraDSs.get(extraField).add((double) extras.get(extraField));
			}
		}

		/**
		 * 
		 * @param dateTime dateTime
		 * @param frequency frequency
		 * @return Bar
		 * @throws Exception Exception
		 */
		public Bar buildBar(LocalDateTime dateTime, Frequency frequency) throws Exception {

			double open = __priceDS.getDouble(0);
			double high = CommonUtils.max(__priceDS);
			double low = CommonUtils.min(__priceDS);
			double close = __priceDS.getDouble(__priceDS.size() - 1);
			double volume = CommonUtils.sum(__volumeDS);
			double amount = CommonUtils.sum(__amountDS);

			return new BasicBar(CommonDataUtils.toMarketDatetime(dateTime), open, high, low, close, volume, close,
			        frequency, MapUtils.putAll(new HashMap<>(), new Object[][] { { "amount", amount } }));
		}

		/*
		 * (non-Javadoc)
		 * @see java.lang.Object#toString()
		 */
		@Override
		public String toString() {

			List<Column<?>> cols = new ArrayList<>();
			cols.add(DateTimeColumn.create(Constants.DATE_TIME, __dateTimes));
			cols.add(DoubleColumn.create(Constants.PRICE, __priceDS.toDoubleArray()));
			cols.add(DoubleColumn.create(Constants.BAR_FIELD_VOLUME, __volumeDS.toDoubleArray()));
			cols.add(DoubleColumn.create(Constants.BAR_FIELD_AMOUNT, __amountDS.toDoubleArray()));
			__extraDSs.forEach((extraField, dal) -> {
				cols.add(DoubleColumn.create(extraField, dal.toDoubleArray()));
			});
			Table df = Table.create(cols);
			return df.toString();
		}
	}

	/**
	 * 
	 * @author qy
	 *
	 */
	private interface WindLiveFeedPoller extends Runnable {

		/**
		 * Is the Poller stopped?
		 * @return stopped
		 */
		boolean stopped();

		/**
		 * Request the Poller stop.
		 */
		void stop();

		/**
		 * set before Thread.start()
		 * 
		 * @param pollInterval pollInterval
		 */
		void setPollInterval(Pair<Long, TimeUnit> pollInterval);

		/**
		 * 获取当前 Bar 的 TimeRange:
		 * 
		 * @return TimeRange
		 */
		TimeRange getCurrentRange();

		/**
		 * <pre>
		 * 返回生成下一个 NEXT_CALL 事件 (Bars 到达) 的时间:
		 * </pre>
		 * 
		 * <pre>
		 * Must return a non-naive datetime.
		 * ZoneId: "UTC"
		 * </pre>
		 * 
		 * @return ZonedDateTime
		 */
		ZonedDateTime getNextCallDateTime();

		/**
		 * 生成 NEXT_CALL 事件: 收集 分笔成交 行情缓存序列中的数据, 生成 Bars, 向 行情事件队列 中插入一个 (ON_BARS, bars) 事件:
		 */
		void doCall();
	}

	/**
	 * <pre>
	 * Producer 线程抽象类: 负责间隔调用 w.wsq() 获取标的 分笔成交 行情, 在周期结束的
	 * NEXT_ALL 事件时, 生成 Bars, 插入 行情事件队列, 并通知相关的 订阅者 (策略):
	 * </pre>
	 * 
	 * <pre>
	 * Producer 线程 实现类:
	 * 1. 每间隔 ActiveWindLivePyFeedPoller._wsqPullInterval 时间, 调用 w.wsq() 获取实时行情数据,
	 * 暂存在 self._tickDSDict 序列集中: WindPollingThread.__wait();
	 * 2. 每间隔 frequency 时间 (在周期 Bar 结束时), 生成一个 (ON_BARS, Bars) 事件, 并插入 self.__queue 队列中; 在生成每个周期 Bars 时,
	 * 读取暂存在 self._tickDSDict 序列集中的 Tick 数据, 汇总得到相应的 Bar (必须保证每次汇总生成 Bar 时, self._tickDSDict 序列只包含当前周期的数据):
	 * ActiveWindLivePyFeedPoller.doCall().
	 * </pre>
	 * 
	 * <pre>
	 * Not using xignite polling thread is because two underscores functions can't be override, e.g.
	 * __wait()
	 * </pre>
	 * 
	 * @author c-geo
	 *
	 */
	private class ActiveWindLiveFeedPoller implements WindLiveFeedPoller {

		private final Frequency __frequency;

		/**
		 * 当前 Bar 的 TimeRange:
		 */
		private TimeRange __currRange;

		/**
		 * ZoneId: "UTC"
		 */
		private ZonedDateTime __nextBarClose;

		private final Map<String, TickDataSeries> __tickDSDict;

		private final Map<String, LocalDateTime> __lastQuoteTimeDict;

		/**
		 * 记录上一次的 Ticks: {identifier: Tick};
		 * lastTick 一定是全量 Tick, 即 field 齐全的 Tick:
		 */
		private final Map<String, Row> __lastTickDict;

		private Pair<Long, TimeUnit> __wsqPollInterval;

		/**
		 * Wapi:
		 */
		private final WapiPrx __wapi;

		private boolean __stopped;

		/**
		 * 
		 * @param wsqPollInterval wsqPollInterval
		 */
		public ActiveWindLiveFeedPoller(Frequency frequency, Pair<Long, TimeUnit> wsqPollInterval) {

			__frequency = frequency;
			__tickDSDict = new HashMap<>();
			__lastQuoteTimeDict = new HashMap<>();
			__lastTickDict = new HashMap<>();
			__wsqPollInterval = wsqPollInterval;

			for (String identifier : __identifiers) {
				__tickDSDict.put(identifier, new TickDataSeries());
			}

			// Create a proxy to the Wapi
			ObjectPrx obj = Services.getInstance().createIceWapiProxy("Wapi");
			// Downcast the proxy to a Wapi proxy
			__wapi = WapiPrx.checkedCast(obj);

			__stopped = false;

			__currRange = null;
			__nextBarClose = null;

			__updateNextBarClose();
		}

		/*
		 * 
		 */
		@Override
		public boolean stopped() {

			return __stopped;
		}

		/*
		 * 
		 */
		@Override
		public void stop() {

			__stopped = true;
		}

		/*
		 * 间隔调用 tushare.get_realtime_quotes() 获取 A 股分笔成交行情, 在周期结束的
		 * NEXT_CALL 事件时, 生成 Bars, 插入 行情事件队列, 并通知相关的 订阅者 (策略):
		 * (non-Javadoc)
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {

			logger.info("ActiveWindLiveFeedPoller thread started.");

			// Wapi.start()
			if (!__wapi.isconnected()) {
				WindData wd = __wapi.start(OptionalLong.of(WAPI_START_TIMEOUT_MS));
				if (wd.errorCode != 0) {
					__stopped = true; // this will cause WindLiveFeed reaching EOF
					logger.error("Start Wind Quant API with error: {}", Helper.toString(wd));
				}
			}

			while (!__stopped) {
				try {
					__wait();
					if (!__stopped) {
						doCall();
					}
				} catch (Exception e) {
					logger.error("Unhandled exception: {}", ExceptionUtils.getStackTrace(e));
				}
			}

			logger.info("ActiveWindLiveFeedPoller thread finished.");
		}

		/**
		 * 
		 */
		private void __updateNextBarClose() {

			__currRange = TimeRange.buildRange(CommonDataUtils.utcNow(), __frequency);
			__nextBarClose = __currRange.getEnding();
		}

		/*
		 * (non-Javadoc)
		 * @see qy.qyalgotrader.wind.barfeed.WindLiveFeed.WindLiveFeedPoller#setPollInterval()
		 */
		@Override
		public void setPollInterval(Pair<Long, TimeUnit> pollInterval) {

			__wsqPollInterval = pollInterval;
		}

		/*
		 * (non-Javadoc)
		 * @see qy.qyalgotrader.wind.barfeed.WindLiveFeed.WindLiveFeedPoller#getCurrentRange()
		 */
		@Override
		public TimeRange getCurrentRange() {

			return __currRange;
		}

		/*
		 * (non-Javadoc)
		 * @see qy.qyalgotrader.wind.barfeed.WindLiveFeed.WindLiveFeedPoller#getNextCallDateTime()
		 */
		@Override
		public ZonedDateTime getNextCallDateTime() {

			return __nextBarClose;
		}

		/*
		 * (non-Javadoc)
		 * @see qy.qyalgotrader.wind.barfeed.WindLiveFeed.WindLiveFeedPoller#doCall()
		 */
		@Override
		public void doCall() {

			// 本次生成 (ON_BARS, Bars) 事件中 Bar 的 时间:
			ZonedDateTime barDateTime = __currRange.getBeginning();
			ZonedDateTime barEndTime = __currRange.getEnding();

			__updateNextBarClose();

			if (__useSseSzseTradeTime) {
				// 非交易时间不生成 Bars:
				if (!CommonDataUtils.isSseSzseTickDatetimeInBarTimeRange(__identifiers.get(0),
						CommonDataUtils.toMarketDatetime(barDateTime).toLocalDateTime(),
						OptionalInt.empty(), Optional.empty(), Optional.empty()) &&
						!CommonDataUtils.isSseSzseTickDatetimeInBarTimeRange(__identifiers.get(0),
								CommonDataUtils.toMarketDatetime(barEndTime).toLocalDateTime(),
								OptionalInt.empty(), Optional.empty(), Optional.empty())) {
					return;
				}
			}

			Map<String, Bar> barDict = new HashMap<>();
			for (String identifier : __identifiers) {
				try {
					if (!__tickDSDict.get(identifier).empty()) {
						if (logger.isDebugEnabled()) {
							logger.debug("Building Bar for {} ({}), ticks: \n{}", identifier,
							        CommonDataUtils.toMarketDatetime(barDateTime), __tickDSDict.get(identifier));
						}
						Bar bar = __tickDSDict.get(identifier).buildBar(
								CommonDataUtils.toMarketDatetime(barDateTime).toLocalDateTime(), __frequency);
						if (logger.isDebugEnabled()) {
							logger.debug("Bar ({}) ok: \n{}", identifier, bar);
						}
						barDict.put(identifier, bar);
					} else {  // empty bar:
						if (__genEmptyBars) {
							Bar lastBar = null;
							Pair<EventType, Object> e = __deQueue.peekLast();
							if (e != null && e.getLeft() == EventType.ON_BARS) {
								Bars bars = (Bars) e.getRight();
								lastBar = bars.getBar(identifier);
							}
							if (lastBar != null && barDateTime.compareTo(lastBar.getDateTime()) > 0) {
								Bar currBar = CommonDataUtils.genEmptyBarBy(CommonDataUtils.toMarketDatetime(barDateTime), lastBar);
								barDict.put(identifier, currBar);
							}
						}
					}
				} catch (Exception e) {
					logger.error("buildBar({}, {}, {}) for {} error: {}", CommonDataUtils.toMarketDatetime(barDateTime),
					        __tickDSDict.get(identifier), __frequency, identifier, ExceptionUtils.getStackTrace(e));
				}
			}

			if (barDict.size() > 0) {
				try {
					Bars bars = new Bars(barDict);
					__deQueue.put(Pair.of(EventType.ON_BARS, bars));
				} catch (Exception e) {
					logger.error("Exception occured: {}", ExceptionUtils.getStackTrace(e));
				}
			}
		}

		/**
		 * 在 NEXT_CALL (产生 ON_BAR 事件) 之前, 每间隔 self._wsqPullInterval, 调用一次 __getTushareTickData():
		 */
		private void __wait() {

			// first reset ticks info in one cycle, maybe we need save it if NO quotation in this period
			for (String identifier : __identifiers) {
				__tickDSDict.get(identifier).reset();
			}

			ZonedDateTime nextCallTime = getNextCallDateTime();

			// TODO: 首次调用, 先 sleep() 到最近的一个标准 Tick 时间 slot:
			/*
			 * main loop: 阻塞直到 nextCallTime
			 */
			ZonedDateTime utcNow = CommonDataUtils.utcNow();
			while (!__stopped && utcNow.compareTo(nextCallTime) < 0) {

				if (logger.isDebugEnabled()) {
					logger.debug("utcnow(): {}, nextCallTime: {}", CommonDataUtils.toMarketDatetime(utcNow),
					        CommonDataUtils.toMarketDatetime(nextCallTime));
				}
				ZonedDateTime startTime = CommonDataUtils.utcNow();
				// pull for wind tick data:
				__getWindTickData();
				ZonedDateTime endTime = CommonDataUtils.utcNow();

				Duration timeDiff = Duration.between(startTime, endTime);
				if (timeDiff.getSeconds() < __wsqPollInterval.getLeft()) {
					long consumedMs = timeDiff.toMillis();
					// maxSleepMs: 若当前时间到 nextCallTime 不足一个 pollInterval:
					long maxSleepMs = Duration.between(endTime, nextCallTime).toMillis();
					long sleepMs = Math.min(__wsqPollInterval.getRight().toMillis(__wsqPollInterval.getLeft()) - consumedMs,
							maxSleepMs); // 防止睡过了头
					logger.debug("Time consumed getting Tick data: {} ms, sleep for {} ms", consumedMs, sleepMs);
					try {
						Thread.sleep(sleepMs);
					} catch (InterruptedException e) {
						logger.error("InterruptedException occured: {}", ExceptionUtils.getStackTrace(e));
					}
				}
				utcNow = CommonDataUtils.utcNow();
			}
		}

		/**
		 * <pre>
		 * 调用一次 w.wsq(), 获取所有订阅标的最新的 实时行情, 并一一插入 self._tickDSDict 相应的 序列缓存 中;
		 * w.wsq() 获取的 volume 和 amount 为当日累积的, 需计算得出 Tick 范围内的:
		 * </pre>
		 */
		private void __getWindTickData() {

			try {
				String codes = String.join(",", __identifiers);
				WindData wd = __wapi.wsq(codes, WSQ_FIELDS_STRING, Optional.empty());
				if (wd.errorCode != 0) {
					logger.error("Call wsq({}, {}) with errors: {}", codes, WSQ_FIELDS_STRING, Helper.toString(wd));
				}

				// ticks for all codes:
				Table df = CommonUtils.makeTable(Arrays.asList(wd.fields));
				for (int i = 0; i < wd.codes.length; i++) {
					Row aRow = df.appendRow();
					for (int j = 0; j < wd.fields.length; j++) {
						aRow.setString(wd.fields[j], wd.data[j][i]);
					}
				}

				for (String identifier : __identifiers) {
					int codeIdx = Arrays.binarySearch(wd.codes, identifier);
					if (codeIdx < 0) {
						throw new IllegalStateException(String.format("identifier %s NOT found in wsq response: %s",
						        identifier, Helper.toString(wd)));
					}
					Row tick = df.row(codeIdx);

					String dtField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_date", "resp");
					String tmField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_time", "resp");
					String priceField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_last", "resp");
					String volField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_vol", "resp");
					String amtField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_amt", "resp");

					if (__validTickDataActive(identifier, tick)) {
						// @formatter:off
						if (__lastTickDict.get(identifier) == null) {
							// TODO:
							__tickDSDict.get(identifier).appendTick(
									Helper.makeDateTime(tick.getString(dtField), tick.getString(tmField)),
									Double.parseDouble(tick.getString(priceField)),
									Double.parseDouble(tick.getString(volField)),
									Double.parseDouble(tick.getString(amtField)));

							logger.debug("Initial Tick ({}) received: \n{}", identifier, tick);
						} else {
							Row lastTick = __lastTickDict.get(identifier);
							__tickDSDict.get(identifier).appendTick(
									Helper.makeDateTime(tick.getString(dtField), tick.getString(tmField)),
									Double.parseDouble(tick.getString(priceField)),
									Double.parseDouble(tick.getString(volField))
											- Double.parseDouble(lastTick.getString(volField)),
									Double.parseDouble(tick.getString(amtField))
											- Double.parseDouble(lastTick.getString(amtField)));

						}
						// @formatter:on
						// 更新记录的上一个有效 Tick:
						__lastTickDict.put(identifier, tick);
					}
				}
			} catch (Exception e) {
				logger.error("Wind polling exception: {}", ExceptionUtils.getStackTrace(e));
			}
		}

		/**
		 * <p>
		 * 1. 确保 tickInfo 中的 time 一定晚于已记录的指定标的最新实时行情时间 (self._lastQuotationTime), 并更新之, 否则返回 Fale (放弃此条
		 * tickInfo);
		 * </p>
		 * <p>
		 * 2. 确保 tickInfo 中的 vol 一定大于已记录的上一个 tickInfo 的 vol, 否则返回 Fale (放弃此条 tickInfo);
		 * </p>
		 * <p>
		 * 3. 确保最新成交价格为合法.
		 * </p>
		 * 
		 * @param identifier identifier
		 * @param tick tick
		 * @return isValidTickData
		 */
		private boolean __validTickDataActive(String identifier, Row tick) {

			String dtField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_date", "resp");
			String tmField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_time", "resp");
			String volField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_vol", "resp");
			String preCloseField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_pre_close", "resp");
			String priceField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_last", "resp");
			LocalDateTime tickDateTime = Helper.makeDateTime(tick.getString(dtField), tick.getString(tmField));

			if (__lastQuoteTimeDict.get(identifier) == null
			        || __lastQuoteTimeDict.get(identifier).compareTo(tickDateTime) < 0) {
				__lastQuoteTimeDict.put(identifier, tickDateTime);
			} else {
				return false;
			}

			if (__lastTickDict.get(identifier) != null
			        && __lastTickDict.get(identifier).getString(volField).equals(tick.getString(volField))) {
				return false;
			}
			// TODO: 精确到具体品种的涨跌停板:
			// @formatter:off
			return !(Double.parseDouble(tick.getString(priceField)) > Double.parseDouble(tick.getString(preCloseField))
					* Constants.PRICE_LMT_CN_STOCK_A_UP) &&
					!(Double.parseDouble(tick.getString(priceField)) < Double.parseDouble(tick.getString(preCloseField))
							* Constants.PRICE_LMT_CN_STOCK_A_DOWN);
			// @formatter:on
		}
	}

	/**
	 * 
	 * @author c-geo
	 *
	 */
	private class PassiveWindLiveFeedPoller implements WindLiveFeedPoller {

		private class WsqCallBackImpl implements WsqCallBack {

			@Override
			public void cbf(WindData windData, Current current) {

				// 在 Ice 线程中接收 wsq() 实时行情返回的 WindData, 并放入 __tickDSDict 相应的 TickDataSeries 中:
				if (windData.errorCode != 0) {
					logger.warn("error wsq() realtime tick data received: {}", Helper.toString(windData));
				} else {
					try {
						if (logger.isDebugEnabled()) {
							logger.debug("wsq() ticks received: {}", Helper.toString(windData));
						}
						// ticks for all received codes:
						Table df = CommonUtils.makeTable(Arrays.asList(windData.fields));
						for (int i = 0; i < windData.codes.length; i++) {
							Row aRow = df.appendRow();
							for (int j = 0; j < windData.fields.length; j++) {
								aRow.setString(windData.fields[j], windData.data[j][i]);
							}
						}
						for (int i = 0; i < windData.codes.length; i++) {
							String identifier = windData.codes[i];
							Row tick = df.row(i);

							String dtField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_date", "resp");
							String tmField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_time", "resp");
							String priceField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_last", "resp");
							String volField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_vol", "resp");
							String amtField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_amt", "resp");

							if (__validTickDataPassive(identifier, tick)) {
								// @formatter:off
								synchronized (__tickDSDict.get(identifier)) {
									if (__lastTickDict.get(identifier) == null) {  // initial Tick received:
										// TODO:
										__tickDSDict.get(identifier).appendTick(
												Helper.makeDateTime(tick.getString(dtField), tick.getString(tmField)),
												Double.parseDouble(tick.getString(priceField)),
												Double.parseDouble(tick.getString(volField)),
												Double.parseDouble(tick.getString(amtField)));

										logger.debug("Initial Tick ({}) received: \n{}", identifier, tick);
										// 直接更新记录的上一个有效 Tick:
										__lastTickDict.put(identifier, tick);
									} else {  // 尽管非 initial Tick, 但 lastTick 也是全量 Tick (field 齐全):
										Row lastTick = __lastTickDict.get(identifier);
										Map<String, String> theTick = CommonUtils.tableRow2MapString(lastTick, Optional.empty());
										for (String col : tick.columnNames()) {
											theTick.put(col, tick.getString(col));
										}
										__tickDSDict.get(identifier).appendTick(
												Helper.makeDateTime(theTick.get(dtField), theTick.get(tmField)),
												Double.parseDouble(theTick.get(priceField)),
												Double.parseDouble(theTick.get(volField))
														- Double.parseDouble(lastTick.getString(volField)),
												Double.parseDouble(theTick.get(amtField))
														- Double.parseDouble(lastTick.getString(amtField)));
										// 使用最新 Tick 合并 last Tick:
										for (String col : tick.columnNames()) {
											lastTick.setString(col, tick.getString(col));
										}
										__lastTickDict.put(identifier, lastTick);
									}
								}
								// @formatter:on
							}
						}
					} catch (Exception e) {
						logger.error("Exception occurred in Ice Callback: {}", ExceptionUtils.getStackTrace(e));
					}
				}
			}
		}

		private final Frequency __frequency;

		/**
		 * 当前 Bar 的 TimeRange:
		 */
		private TimeRange __currRange;

		/**
		 * ZoneId: "UTC"
		 */
		private ZonedDateTime __nextBarClose;

		private final Map<String, TickDataSeries> __tickDSDict;

		private final Map<String, LocalDateTime> __lastQuoteTimeDict;

		/**
		 * 记录上一次的 Ticks: {identifier: Tick};
		 * lastTick 一定是全量 Tick, 即 field 齐全的 Tick:
		 */
		private final Map<String, Row> __lastTickDict;

		private Pair<Long, TimeUnit> __wsqPollInterval;

		/**
		 * Wapi:
		 */
		private final WapiPrx __wapi;

		/**
		 * WsqCallBack 本地实现:
		 */
		private final WsqCallBack __wsqCallBack;

		/**
		 * WsqCallBack 的本地代理:
		 */
		private final WsqCallBackPrx __wsqCallBackPrx;

		private boolean __stopped;

		/**
		 * 
		 * @param frequency frequency
		 * @param wsqPollInterval wsqPollInterval
		 */
		public PassiveWindLiveFeedPoller(Frequency frequency, Pair<Long, TimeUnit> wsqPollInterval) {

			__frequency = frequency;
			__tickDSDict = new HashMap<>();
			__lastQuoteTimeDict = new HashMap<>();
			__lastTickDict = new HashMap<>();
			__wsqPollInterval = wsqPollInterval;

			for (String identifier : __identifiers) {
				__tickDSDict.put(identifier, new TickDataSeries());
			}

			// Create a proxy to the Wapi:
			ObjectPrx obj = Services.getInstance().createIceWapiProxy("Wapi");
			// Downcast the proxy to a Wapi proxy:
			__wapi = WapiPrx.checkedCast(obj);
			// create and bind the WsqCallBack:
			__wsqCallBack = new WsqCallBackImpl();
			ObjectAdapter adapter = Services.getInstance().getCommunicator().createObjectAdapter("");
			__wsqCallBackPrx = WsqCallBackPrx.checkedCast(adapter.addWithUUID(__wsqCallBack));
			__wapi.ice_getCachedConnection().setAdapter(adapter);
			__wapi.ice_getCachedConnection().setACM(OptionalInt.empty(), Optional.empty(), Optional.of(ACMHeartbeat.HeartbeatAlways));
			adapter.activate();

			__stopped = false;

			__currRange = null;
			__nextBarClose = null;

			__updateNextBarClose();
		}

		/*
		 * 
		 */
		@Override
		public boolean stopped() {

			return __stopped;
		}

		/*
		 * 
		 */
		@Override
		public void stop() {

			__stopped = true;
		}

		/*
		 * 使用 wsq() 异步订阅模式, 获取 A 股分笔成交行情, 在周期结束的
		 * NEXT_CALL 事件时, 生成 Bars, 插入 行情事件队列, 并通知相关的 订阅者 (策略):
		 * (non-Javadoc)
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {

			logger.info("PassiveWindLiveFeedPoller thread started.");

			// Wapi.start()
			if (!__wapi.isconnected()) {
				WindData wd = __wapi.start(OptionalLong.of(WAPI_START_TIMEOUT_MS));
				if (wd.errorCode != 0) {
					__stopped = true; // this will cause WindLiveFeed reaching EOF
					logger.error("Start Wind Quant API with error: {}", Helper.toString(wd));
				}
			}

			// 添加 wsq() 异步 tick 行情订阅:
			String codes = String.join(",", __identifiers);
			WindData wd = __wapi.wsqSubscribe(codes, WSQ_FIELDS_STRING, __wsqCallBackPrx, Optional.empty());
			if (wd.errorCode != 0) {
				// wsq() tick 实时行情订阅出错, 则直接结束:
				logger.error("wsq() subscribe error: {}", Helper.toString(wd));
				__stopped = true;
			}
			logger.info("wsq() subscribe windcodes: {} (total: {}), subscription info: {}", __identifiers,
					__identifiers.size(), Helper.toString(wd));

			while (!__stopped) {
				try {
					__wait();
					if (!__stopped) {
						doCall();
					}
				} catch (Exception e) {
					logger.error("Unhandled exception: {}", ExceptionUtils.getStackTrace(e));
				}
			}

			logger.info("PassiveWindLiveFeedPoller thread finished.");
		}

		/**
		 *
		 */
		private void __updateNextBarClose() {

			__currRange = TimeRange.buildRange(CommonDataUtils.utcNow(), __frequency);
			__nextBarClose = __currRange.getEnding();
		}

		/*
		 * (non-Javadoc)
		 * @see qy.qyalgotrader.wind.barfeed.WindLiveFeed.WindLiveFeedPoller#setPollInterval()
		 */
		@Override
		public void setPollInterval(Pair<Long, TimeUnit> pollInterval) {

			__wsqPollInterval = pollInterval;
		}

		/*
		 * (non-Javadoc)
		 * @see qy.qyalgotrader.wind.barfeed.WindLiveFeed.WindLiveFeedPoller#getCurrentRange()
		 */
		@Override
		public TimeRange getCurrentRange() {

			return __currRange;
		}

		/*
		 * (non-Javadoc)
		 * @see qy.qyalgotrader.wind.barfeed.WindLiveFeed.WindLiveFeedPoller#getNextCallDateTime()
		 */
		@Override
		public ZonedDateTime getNextCallDateTime() {

			return __nextBarClose;
		}

		/*
		 * (non-Javadoc)
		 * @see qy.qyalgotrader.wind.barfeed.WindLiveFeed.WindLiveFeedPoller#doCall()
		 */
		@Override
		public void doCall() {

			// 本次生成 (ON_BARS, Bars) 事件中 Bar 的 时间:
			ZonedDateTime barDateTime = __currRange.getBeginning();
			ZonedDateTime barEndTime = __currRange.getEnding();

			__updateNextBarClose();

			if (__useSseSzseTradeTime) {
				// 非交易时间不生成 Bars:
				if (!CommonDataUtils.isSseSzseTickDatetimeInBarTimeRange(__identifiers.get(0),
						CommonDataUtils.toMarketDatetime(barDateTime).toLocalDateTime(),
						OptionalInt.empty(), Optional.empty(), Optional.empty()) &&
						!CommonDataUtils.isSseSzseTickDatetimeInBarTimeRange(__identifiers.get(0),
								CommonDataUtils.toMarketDatetime(barEndTime).toLocalDateTime(),
								OptionalInt.empty(), Optional.empty(), Optional.empty())) {
					return;
				}
			}

			Map<String, Bar> barDict = new HashMap<>();
			for (String identifier : __identifiers) {
				synchronized (__tickDSDict.get(identifier)) {
					try {
						if (!__tickDSDict.get(identifier).empty()) {
							if (logger.isDebugEnabled()) {
								logger.debug("Building Bar for {} ({}), ticks: \n{}", identifier,
										CommonDataUtils.toMarketDatetime(barDateTime), __tickDSDict.get(identifier));
							}
							Bar bar = __tickDSDict.get(identifier).buildBar(
									CommonDataUtils.toMarketDatetime(barDateTime).toLocalDateTime(), __frequency);
							if (logger.isDebugEnabled()) {
								logger.debug("Bar ({}) ok: \n{}", identifier, bar);
							}
							barDict.put(identifier, bar);
						} else {  // empty bar:
							if (__genEmptyBars) {
								Bar lastBar = null;
								Pair<EventType, Object> e = __deQueue.peekLast();
								if (e != null && e.getLeft() == EventType.ON_BARS) {
									Bars bars = (Bars) e.getRight();
									lastBar = bars.getBar(identifier);
								}
								if (lastBar != null) {
									Bar currBar = CommonDataUtils.genEmptyBarBy(CommonDataUtils.toMarketDatetime(barDateTime), lastBar);
									barDict.put(identifier, currBar);
								}
							}
						}
					} catch (Exception e) {
						logger.error("buildBar({}, {}, {}) for {} error: {}", CommonDataUtils.toMarketDatetime(barDateTime),
								__tickDSDict.get(identifier), __frequency, identifier, ExceptionUtils.getStackTrace(e));
					}
				}
			}

			if (barDict.size() > 0) {
				try {
					Bars bars = new Bars(barDict);
					__deQueue.put(Pair.of(EventType.ON_BARS, bars));
				} catch (Exception e) {
					logger.error("Exception occured: {}", ExceptionUtils.getStackTrace(e));
				}
			}
		}

		/**
		 * 在 NEXT_CALL (产生 ON_BAR 事件) 之前, 保持阻塞 (tick 行情采用异步订阅模式接收):
		 */
		private void __wait() {

			// first reset ticks info in one cycle, maybe we need save it if NO quotation in this period
			for (String identifier : __identifiers) {
				synchronized (__tickDSDict.get(identifier)) {
					__tickDSDict.get(identifier).reset();
				}
			}

			ZonedDateTime nextCallTime = getNextCallDateTime();

			// TODO: 首次调用, 先 sleep() 到最近的一个标准 Tick 时间 slot:
			/*
			 * main loop: 阻塞直到 nextCallTime
			 */
			ZonedDateTime utcNow = CommonDataUtils.utcNow();
			while (!__stopped && utcNow.compareTo(nextCallTime) < 0) {
				if (logger.isDebugEnabled()) {
					logger.debug("utcnow(): {}, nextCallTime: {}", CommonDataUtils.toMarketDatetime(utcNow),
							CommonDataUtils.toMarketDatetime(nextCallTime));
				}

				// maxSleepMs: 若当前时间到 nextCallTime 不足一个 pollInterval:
				long maxSleepMs = Duration.between(utcNow, nextCallTime).toMillis();
				long sleepMs = Math.min(__wsqPollInterval.getRight().toMillis(__wsqPollInterval.getLeft()), maxSleepMs); // 防止睡过了头
				try {
					Thread.sleep(sleepMs);
				} catch (InterruptedException e) {
					logger.error("InterruptedException occured: {}", ExceptionUtils.getStackTrace(e));
				}

				utcNow = CommonDataUtils.utcNow();
			}
		}

		/**
		 * <p>
		 * 1. 确保 tickInfo 中的 time 一定晚于已记录的指定标的最新实时行情时间 (self._lastQuotationTime), 并更新之, 否则返回 Fale (放弃此条
		 * tickInfo);
		 * </p>
		 * <p>
		 * 2. 确保 tickInfo 中的 vol 一定大于已记录的上一个 tickInfo 的 vol, 否则返回 Fale (放弃此条 tickInfo);
		 * </p>
		 * <p>
		 * 3. 确保最新成交价格为合法.
		 * </p>
		 *
		 * @param identifier identifier
		 * @param tick tick
		 * @return isValidTickData
		 */
		private boolean __validTickDataPassive(String identifier, Row tick) {

			String dtField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_date", "resp");
			String tmField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_time", "resp");
			String volField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_vol", "resp");
			String preCloseField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_pre_close", "resp");
			String priceField = (String) Constants.WIND_FIELD_NAME_MAPPING_RT_QUOTES.get("rt_last", "resp");
			Map<String, String> theTick = null;  // Map<String, String> 结构的 Tick 方便进行合并处理
			if (__lastTickDict.get(identifier) == null) {  // 首个 Tick, 要求 field 齐全:
				if (WSQ_FIELDS.size() != tick.columnNames().size()) {
					logger.error("Illegal initial Tick: {} with incomplete fields!", tick);
					return false;
				}
				theTick = CommonUtils.tableRow2MapString(tick, Optional.empty());
			} else {  // 接收到增量 Tick, 合并到上一个 Tick:
				Row lastTick = __lastTickDict.get(identifier);
				theTick = CommonUtils.tableRow2MapString(lastTick, Optional.empty());
				// 合并增量 Tick:
				for (String col : tick.columnNames()) {
					theTick.put(col, tick.getString(col));
				}
			}

			LocalDateTime tickDateTime = Helper.makeDateTime(theTick.get(dtField), theTick.get(tmField));

			if (__lastQuoteTimeDict.get(identifier) == null
					|| __lastQuoteTimeDict.get(identifier).compareTo(tickDateTime) < 0) {
				__lastQuoteTimeDict.put(identifier, tickDateTime);
			} else {
				return false;
			}

			if (__lastTickDict.get(identifier) != null
					&& __lastTickDict.get(identifier).getString(volField).equals(theTick.get(volField))) {
				return false;
			}
			// TODO: 精确到具体品种的涨跌停板:
			// @formatter:off
			return !(Double.parseDouble(theTick.get(priceField)) > Double.parseDouble(theTick.get(preCloseField))
					* Constants.PRICE_LMT_CN_STOCK_A_UP) &&
					!(Double.parseDouble(theTick.get(priceField)) < Double.parseDouble(theTick.get(preCloseField))
							* Constants.PRICE_LMT_CN_STOCK_A_DOWN);
			// @formatter:on
		}
	}

	/**
	 * 必须是 WindCode:
	 */
	private final List<String> __identifiers;

	private final Path __hisTickCsvBase;

	private final Path __todayTickCsvBase;

	private final Path __hisMinCsvBase;

	/**
	 * 是否生成成交量 (额) 为 0 的 Bar:
	 */
	private final boolean __genEmptyBars;

	/**
	 * WindLiveFeedPoller#doCall() 中是否按照 沪深交易所 A 股交易时间生成 Bars:
	 */
	private final boolean __useSseSzseTradeTime;

	/**
	 * wsq 获取行情是否使用 主动模式 (非订阅模式, 无需 callback):
	 */
	private final boolean __useActiveMode;

	/**
	 * self.__queue: 行情事件队列, 存放 (TuShareBarFeedThread.ON_BARS, Bars):
	 */
	private final BlockingDeque<Pair<EventType, Object>> __deQueue;

	private final WindLiveFeedPoller __poller;

	private final Thread __thread;

	/**
	 * 
	 * @param identifiers identifiers
	 * @param frequency frequency
	 * @param maxLenP default: 0
	 * @param replayDaysP default: -1
	 * @param hisTickCsvBaseP default: null
	 * @param todayTickCsvBaseP default: null
	 * @param hisMinCsvBaseP default: WindDataUtils.WIND_HIS_DATA_BASE_DIR_DEFAULT
	 * @param wsqPollIntervalP default: WIND_FUTURES_INQUERY_PERIOD_DEFAULT
	 * @param genEmptyBarsP default: true
	 * @param useSseSzseTradeTimeP default: false
	 * @param useActiveModeP default: false
	 * @throws Exception Exception
	 */
	// @formatter:off
	public WindLiveFeed(List<String> identifiers, Frequency frequency,
	                    OptionalInt maxLenP, OptionalInt replayDaysP,
	                    Optional<Path> hisTickCsvBaseP,
	                    Optional<Path> todayTickCsvBaseP,
	                    Optional<Path> hisMinCsvBaseP,
	                    Optional<Pair<Long, TimeUnit>> wsqPollIntervalP,
			            Optional<Boolean> genEmptyBarsP, Optional<Boolean> useSseSzseTradeTimeP,
			            Optional<Boolean> useActiveModeP) throws Exception {

		this(identifiers, frequency,
				maxLenP.orElse(0),
				replayDaysP.orElse(-1),
				hisTickCsvBaseP.orElse(null),
				todayTickCsvBaseP.orElse(null),
				hisMinCsvBaseP.orElse(WindDataUtils.WIND_HIS_DATA_BASE_DIR_DEFAULT),
				wsqPollIntervalP.orElse(WIND_FUTURES_INQUERY_PERIOD_DEFAULT),
				genEmptyBarsP.orElse(true),
				useSseSzseTradeTimeP.orElse(false),
				useActiveModeP.orElse(false));
	}
	// @formatter:on

	/**
	 * 
	 * @param identifiers identifiers
	 * @param frequency frequency
	 * @param maxLen maxLen
	 * @param replayDays replayDays
	 * @param hisTickCsvBase hisTickCsvBase
	 * @param todayTickCsvBase todayTickCsvBase
	 * @param hisMinCsvBase hisMinCsvBase
	 * @param wsqPollInterval wsqPollInterval
	 * @param genEmptyBars genEmptyBars
	 * @param useActiveMode useActiveMode
	 * @throws Exception Exception
	 */
	public WindLiveFeed(List<String> identifiers, Frequency frequency, int maxLen, int replayDays, Path hisTickCsvBase,
	                    Path todayTickCsvBase, Path hisMinCsvBase, Pair<Long, TimeUnit> wsqPollInterval,
	                    boolean genEmptyBars, boolean useSseSzseTradeTime, boolean useActiveMode) throws Exception {

		super(frequency, maxLen);

		__identifiers = identifiers;
		// 历史 / 当日历史 Tick, 历史 minute 行情 csv 文件根路径:
		__hisTickCsvBase = hisTickCsvBase;
		__todayTickCsvBase = todayTickCsvBase;
		__hisMinCsvBase = hisMinCsvBase;

		__genEmptyBars = genEmptyBars;
		__useSseSzseTradeTime = useSseSzseTradeTime;
		__useActiveMode = useActiveMode;

		__deQueue = new LinkedBlockingDeque<>();

		__fillTodayAndHistoryBars(replayDays); // should run before polling thread start

		if (__useActiveMode) {
			__poller = new ActiveWindLiveFeedPoller(frequency, wsqPollInterval);
		} else {
			__poller = new PassiveWindLiveFeedPoller(frequency, wsqPollInterval);
		}
		__thread = new Thread(__poller);
		for (String instrument : __identifiers) {
			registerInstrument(instrument);
		}
	}

	// barfeed.BaseBarFeed interface

	/* (non-Javadoc)
	 * @see qy.jalgotrade.barfeed.BaseBarFeed#getCurrentDateTime()
	 */
	@Override
	public ZonedDateTime getCurrentDateTime() {

//		return CommonDataUtils.utcNow();
		Bars currBars = getCurrentBars();
		if (currBars != null) {
			return currBars.getDateTime();
		} else {
			return CommonDataUtils.utcNow();
		}
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.barfeed.BaseBarFeed#barsHaveAdjClose()
	 */
	@Override
	public boolean barsHaveAdjClose() {

		return false;
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.barfeed.BaseBarFeed#getNextBars()
	 */
	@Override
	public Bars getNextBars() throws Exception {

		Bars ret = null;
		Pair<EventType, Object> item = __deQueue.poll(QUEUE_POLL_TIMEOUT.getLeft(), QUEUE_POLL_TIMEOUT.getRight());
		if (item != null) {
			if (item.getLeft() == EventType.ON_BARS) {
				ret = (Bars) item.getRight();
			} else {
				logger.error("Invalid event received: {} - {} !", item.getLeft(), item.getRight());
			}
			// 若 queue.get() 超时, 则返回 None:
		}
		return ret;
	}

	// observer.Subject interface

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#start()
	 */
	@Override
	public void start() {

		if (__thread.isAlive()) {
			throw new IllegalStateException("Already strated!");
		}
		// Start the thread that runs the client.
		__thread.start();
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#stop()
	 */
	@Override
	public void stop() {

		// __thread.stop()
		__poller.stop();
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#join()
	 */
	@Override
	public void join() {

		if (__thread.isAlive()) {
			try {
				__thread.join();
			} catch (InterruptedException e) {
				logger.warn("InterruptedException: {}", ExceptionUtils.getStackTrace(e));
			}
		}
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#eof()
	 */
	@Override
	public boolean eof() {

		// return self.__thread.stopped()
		return __poller.stopped();
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#peekDateTime()
	 */
	@Override
	public ZonedDateTime peekDateTime() {

		return null;
	}

	// WindLiveFeed own interface
	// TODO
	/**
	 * 
	 * @param pollInterval pollInterval
	 */
	public void setPollInterval(Pair<Long, TimeUnit> pollInterval) {

		if (!__thread.isAlive()) {
			__poller.setPollInterval(pollInterval);
		} else {
			throw new IllegalStateException("can NOT set pollInterval when the poller thread is running.");
		}
	}

	/**
	 * <pre>
	 * replayDays: 要获取的历史行情天数 (不包括当前交易日), 以支持历史行情重放:
	 *     -1 (或 <0): 不获取历史行情;
	 *     0: 获取当前交易日已经发生的行情;
	 *     >0: 要获取的历史行情的天数.
	 * </pre>
	 * 
	 * @param replayDays replayDays
	 * @throws Exception Exception
	 */
	private void __fillTodayAndHistoryBars(int replayDays) throws Exception {

		LocalDate today = CommonDataUtils.nowTradingDate();
		if (replayDays < 0) {
			replayDays = -1; // only allow -1 and >=0 integer value
		}
		if (replayDays == -1) {
			// 纯实时行情模式, 仅接收新行情
		} else if (replayDays == 0) {
			// replay today's quotation
			if (__todayTickCsvBase == null) {
				// tushare 接口调用方式获取当日历史 Tick 行情数据:
				// TODO self.__fillTodayBars()
			} else {
				// csv 文件方式获取当日历史 Tick 行情数据:
				Path todayTickCsvBasePath = __todayTickCsvBase.resolve(today.format(Constants.DATE_FORMAT_YYYY))
						.resolve(today.format(Constants.DATE_FORMAT_YYYYMMDD));
				// TODO self.__addBarsFromCSV(today_, todayTickCsvBasePath)
			}
		} else {
			// 获取指定交易天数的历史行情, 和当前交易日已发生的行情, 使得 Feed 支持对其进行重放:
			if (Frequency.getIntradayFreqsCommon().contains(getFrequency())) {
				// XXX: 日内分钟频, 尝试读取历史 minute 行情:
				if (__hisMinCsvBase == null) {
					// 接口调用方式获取当日历史 minute 行情数据:
				} else {
					// csv 文件方式获取历史 minute 行情数据:
					for (LocalDate td : WindDataUtils.getTradingDays(today, replayDays)) {
						__addBarsFromMinutesCsv(td, __hisMinCsvBase, __hisTickCsvBase, ".csv", "UTF-8");
					}
				}
			}
			/*if (__hisTickCsvBase == null) {
				// tushare 接口调用方式获取历史 Tick 行情数据:
				// TODO self.__fillHistoryBars(replayDays)
			} else {
				// csv 文件方式获取历史 Tick 行情数据:
				for (LocalDate td : WindDataUtils.getTradingDays(today, replayDays)) {
					Path aHisTickCsvBasePath = __hisTickCsvBase.resolve(td.format(Constants.DATE_FORMAT_YYYY))
							.resolve(td.format(Constants.DATE_FORMAT_YYYYMMDD));
					// TODO self.__addBarsFromCSV(td, aHistoryTickCsvBasePath)
				}
			}*/
			if (__todayTickCsvBase == null) {
				// tushare 接口调用方式获取当日历史 Tick 行情数据:
				// TODO self.__fillTodayBars()
			} else {
				// csv 文件方式获取当日历史 Tick 行情数据:
				Path todayTickCsvBasePath = __todayTickCsvBase.resolve(today.format(Constants.DATE_FORMAT_YYYY))
						.resolve(today.format(Constants.DATE_FORMAT_YYYYMMDD));
				// TODO self.__addBarsFromCSV(today_, todayTickCsvBasePath)
			}
		}
	}

	/**
	 * 
	 * @param theDate theDate
	 * @param csvBase csvBase
	 * @param tickCsvBase tickCsvBase
	 * @param csvFileNameExt csvFileNameExt
	 * @param csvFileEnc csvFileEnc
	 * @throws Exception Exception
	 */
	private void __addBarsFromMinutesCsv(LocalDate theDate, Path csvBase, Path tickCsvBase, String csvFileNameExt,
			String csvFileEnc) throws Exception {

		Map<String, List<Bar>> barsDict = new HashMap<>();
		for (String e : __identifiers) {
			// 一一读取指定交易日的 (分钟级别: min_1, min_3, min_5, ...) 历史行情 csv:

			List<Bar> barList = null;
			try {
				barList = __readBarsFromMinutesCsv(e, theDate, csvBase, tickCsvBase, csvFileNameExt, csvFileEnc);
			} catch (FileNotFoundException ex) {
				logger.warn(ex.getMessage());
			} finally {
				if (barList == null || barList.size() == 0) {
					barList = __readBarsFromTicksCsv(e, theDate, csvBase, tickCsvBase, csvFileNameExt, csvFileEnc);
				}
			}

			if (barList.size() > 0) { // 未从历史行情 csv 中获取 Bar, 则不进行 补线
				barsDict.put(e, barList);
			}
		}
		__fillBars(theDate, barsDict);
	}

	/**
	 * 
	 * @param theDate theDate
	 * @param tickCsvBase tickCsvBase
	 * @param csvFileNameExt csvFileNameExt
	 * @param csvFileEnc csvFileEnc
	 * @throws Exception Exception
	 */
	private void __addBarsFromTickCsv(LocalDate theDate, Path tickCsvBase, String csvFileNameExt,
			String csvFileEnc) throws Exception {

	}

	/**
	 * 向 行情事件队列 中插入 (ON_BARS, bars) 事件, 实现历史行情重放 (对 WindLiveFeed 进行迭代获取 行情 时, 将会依次取出):
	 * <br/>
	 * theBarsDict: {证券代码: 特定交易日的历史行情 barList}
	 * @param theBarsDict 其中 List&lt;Bar&gt; 均未非空
	 * @throws Exception Exception
	 */
	private void __fillBars(LocalDate theDate, Map<String, List<Bar>> theBarsDict) throws Exception {

		Map<String, Integer> allIdxesEnd = new HashMap<>(); // 所有 barList 的长度
		Map<String, Integer> currIdxes = new HashMap<>(); // 当前 index
		Map<String, ZonedDateTime> currDmColl = new HashMap<>(); // 当前 value (ZonedDateTime)
		for (String e : theBarsDict.keySet()) {
			allIdxesEnd.put(e, theBarsDict.get(e).size());
			currIdxes.put(e, 0);
			currDmColl.put(e, theBarsDict.get(e).get(currIdxes.get(e)).getDateTime());
		}
		ZonedDateTime currDm = Collections.min(currDmColl.values());

		Map<String, Bar> currBarDict = new HashMap<>();
		// 停止迭代条件: 所有 barList 的 index 均 reached end:
		Predicate<Object> cond = (obj) -> {
			for (Map.Entry<String, Integer> e : currIdxes.entrySet()) {
				if (e.getValue() < allIdxesEnd.get(e.getKey())) {
					return true;
				}
			}
			return false;
		};
		while (cond.test(null)) {
			for (Map.Entry<String, ZonedDateTime> e : currDmColl.entrySet()) {
				if (e.getValue().equals(currDm)) {
					String identifier = e.getKey();
					if (currIdxes.get(identifier) < allIdxesEnd.get(identifier)) {
						currBarDict.put(identifier, theBarsDict.get(identifier).get(currIdxes.get(identifier)));
						// currIdxes 中相应的 index 后移:
						currIdxes.put(identifier, currIdxes.get(identifier) + 1);
					}
				}
			}
			// 更新 currDmColl 以及 currDm:
			currDmColl.clear();
			for (Map.Entry<String, Integer> e : currIdxes.entrySet()) {
				String identifier = e.getKey();
				if (currIdxes.get(identifier) < allIdxesEnd.get(identifier)) {
					currDmColl.put(identifier, theBarsDict.get(identifier).get(e.getValue()).getDateTime());
				}
			}
			currDm = Collections.min(currDmColl.values());

			Bars bars = new Bars(currBarDict);
			__deQueue.put(Pair.of(EventType.ON_BARS, bars));
			currBarDict.clear();
		}
	}

	private List<Bar> __readBarsFromMinutesCsv(String identifier, LocalDate theDate, Path csvBase, Path tickCsvBase, String csvFileNameExt,
			String csvFileEnc)  throws Exception {

		List<Bar> barList = new ArrayList<>();

		String exchId = Constants.exchSffxToName(CommonDataUtils.windCodeTuple(identifier).getRight());
		String dtStr = theDate.format(Constants.DATE_FORMAT_YYYYMMDD);
		String dtYearStr = theDate.format(Constants.DATE_FORMAT_YYYY);
		Path aCsvFileP = csvBase.resolve(CommonDataUtils.barFreqToHisDataDir(getFrequency())).resolve(dtYearStr)
		        .resolve(dtStr).resolve(identifier + csvFileNameExt);
		Map<String, Object> tdInfos = CommonDataUtils.getTradeInfos(identifier, theDate);

		if (aCsvFileP.toFile().exists()) {

			Table df;
			try (InputStreamReader reader = new InputStreamReader(new FileInputStream(aCsvFileP.toFile()),
			        Charset.forName(csvFileEnc))) {
				// @formatter:off
				Map<String, ColumnType> colsTypeMap = MapUtils.putAll(new HashMap<>(), new Object[][] {
					{ Constants.DATE_TIME, ColumnType.LOCAL_DATE_TIME },
					{ Constants.BAR_FIELD_OPEN, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_CLOSE, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_HIGH, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_LOW, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_VOLUME, ColumnType.DOUBLE },
					{ Constants.BAR_FIELD_AMOUNT, ColumnType.DOUBLE }
				});
				ColumnType[] types = CommonUtils.genColumnsInfoFor(aCsvFileP, colsTypeMap, csvFileEnc);
				// @formatter:on
				CsvReadOptions options = CsvReadOptions.builder(reader).missingValueIndicator("").columnTypes(types)
				        .dateTimeFormat(Constants.DATE_TIME_FORMAT_STD).build();

				df = Table.read().usingOptions(options);
			}

			Mean meanUtils = new Mean();
			double[] closes = df.doubleColumn(Constants.BAR_FIELD_CLOSE).asDoubleArray();
			double[] vols = df.doubleColumn(Constants.BAR_FIELD_VOLUME).asDoubleArray();
			double[] amts = df.doubleColumn(Constants.BAR_FIELD_AMOUNT).asDoubleArray();
			for (int i = 0; i < df.rowCount(); i++) {
				/*
				 * Bar 均价 (vwap) = amount / volume / tradeUnit:
				 *     1. 沪深交易所指数: 不可用;
				 *     2. 郑商所 wsi min_1 数据: 不可用 (0.0, 由于 amount 为 0).
				 * 日内分时均价 (vwap_intraday):
				 *     1. 沪深交易所指数: close 加权平均 (按 amount 加权);
				 *     2. 郑商所 wsi min_1 数据: close 加权平均 (按 volume 加权).
				 * 
				 * TODO: 仅 min_1 频率 Bar 计算 vwap_intraday:
				 */
				Row aRow = df.row(i);
				ZonedDateTime dm = CommonDataUtils.toMarketDatetime(aRow.getDateTime(Constants.DATE_TIME));
				double open = aRow.getDouble(Constants.BAR_FIELD_OPEN);
				double high = aRow.getDouble(Constants.BAR_FIELD_HIGH);
				double low = aRow.getDouble(Constants.BAR_FIELD_LOW);
				double close = aRow.getDouble(Constants.BAR_FIELD_CLOSE);
				double vol = aRow.getDouble(Constants.BAR_FIELD_VOLUME);
				double amt = aRow.getDouble(Constants.BAR_FIELD_AMOUNT);
				// Bar VWAP = amount / volume / tradeUnit (郑商所计算值为 0.0):
				double vwap = Double.NaN;
				double vwapIntraday = Double.NaN;
				if (!DoubleMath.fuzzyEquals(vol, 0.0, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH)) {
					vwap = amt / vol / (double) tdInfos.get("tradeUnit");
				}
				if (Arrays.asList("SSE", "SZSE").contains(exchId) && CommonDataUtils.isSseSzseIndex(identifier)) {
					// 沪深交易所指数 Bar vwap 和 vwap_intraday:
					// vwap: NaN
					// vwap_intraday: close 加权平均 (按 amount 加权):
					vwapIntraday = meanUtils.evaluate(closes, amts, 0, i + 1);
				} else if (exchId.equals("CZCE")) {
					// 郑商所 Bar vwap 和 vwap_intraday:
					// vwap: 0.0
					// vwap_intraday: close 加权平均 (按 volume 加权):
					vwapIntraday = meanUtils.evaluate(closes, vols, 0, i + 1);
				} else {
					double amtAccu = StatUtils.sum(amts, 0, i + 1);
					double volAccu = StatUtils.sum(vols, 0, i + 1);
					if (!DoubleMath.fuzzyEquals(volAccu, 0.0, CommonUtils.FUZZY_EQUAL_TOLERANCE_MATH)) {
						vwapIntraday = amtAccu / volAccu / (double) tdInfos.get("tradeUnit");
					}
				}

				// @formatter:off
				Bar aBar = new BasicBar(dm, open, high, low, close, vol, close, getFrequency(),
				        MapUtils.putAll(new HashMap<>(), new Object[][] {
				        	{ Constants.BAR_FIELD_AMOUNT, amt },
				            { Constants.BAR_FIELD_VWAP, vwap },
				            { Constants.BAR_FIELD_VWAP_INTRADAY, vwapIntraday },
				        }));
				// @formatter:on
				barList.add(aBar);
			}

			if (barList.size() == 0) {
				logger.warn("EMPTY history minutes ({}) csv for {} !", getFrequency(), identifier);
			}

			return barList;
		} else {
			throw new FileNotFoundException(String.format("history minutes (%s) csv for %s does NOT exists! loading bars from tick csv...",
					getFrequency(), identifier));
		}
	}

	/**
	 * 
	 * @param identifier identifier
	 * @param theDate theDate
	 * @param csvBase csvBase
	 * @param tickCsvBase tickCsvBase
	 * @param csvFileNameExt csvFileNameExt
	 * @param csvFileEnc csvFileEnc
	 * @return __readBarsFromTicksCsv
	 * @throws Exception Exception
	 */
	private List<Bar> __readBarsFromTicksCsv(String identifier, LocalDate theDate, Path csvBase, Path tickCsvBase, String csvFileNameExt,
			String csvFileEnc)  throws Exception {

		// TODO:
		return null;
	}
}
